package com.millstein.realtime.app.dwd.log;

import com.alibaba.fastjson.JSONObject;
import com.millstein.realtime.app.base.BaseAppV1;
import com.millstein.realtime.common.Constants;
import com.millstein.realtime.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.PatternTimeoutFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.List;
import java.util.Map;

/**
 * @Description
 * @Author tsing
 * @Date 2024-09-30 16:18
 */
public class Dwd_03_TrafficUserJumpDetail extends BaseAppV1 {

    public static void main(String[] args) {
        new Dwd_03_TrafficUserJumpDetail().init(
                3003,
                3,
                "Dwd_03_TrafficUserJumpDetail",
                Constants.TOPIC_DWD_TRAFFIC_PAGE
        );
    }

    /**
     * 具体数据处理的逻辑，由子类编写
     *
     * @param env          执行环境对象
     * @param streamSource 待处理的数据流
     */
    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> streamSource) {
        // 处理原始流，将其加入事件时间
        SingleOutputStreamOperator<JSONObject> eventStreamSource = streamSource
                .map(JSONObject::parseObject)
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                                    @Override
                                    public long extractTimestamp(JSONObject element, long recordTimestamp) {
                                        return element.getLong("ts");
                                    }
                                }));

        // 定义模式
        Pattern<JSONObject, JSONObject> pattern = Pattern
                .<JSONObject>begin("entry")
                .where(new SimpleCondition<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject value) throws Exception {
                        return value.getJSONObject("page").get("last_page_id") == null;
                    }
                })
                .next("repeat")
                .where(new SimpleCondition<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject value) throws Exception {
                        return value.getJSONObject("page").get("last_page_id") == null;
                    }
                })
                .within(Time.seconds(5));

        // 将模式作用到流上，得到模式流
        PatternStream<JSONObject> cepStream = CEP.pattern(eventStreamSource, pattern);

        // 从模式流中获取到数据
        OutputTag<String> timeoutTag = new OutputTag<String>("timeout") {};
        SingleOutputStreamOperator<String> resultStream = cepStream.select(
                // 超时标签
                timeoutTag,
                // 超时处理类
                new PatternTimeoutFunction<JSONObject, String>() {
                    @Override
                    public String timeout(
                            Map<String, List<JSONObject>> pattern, long timeoutTimestamp
                    ) throws Exception {
                        return pattern.get("entry").get(0).toJSONString();
                    }
                },
                // 正常处理类
                new PatternSelectFunction<JSONObject, String>() {
                    @Override
                    public String select(Map<String, List<JSONObject>> pattern) throws Exception {
                        return pattern.get("entry").get(0).toJSONString();
                    }
                });

        // 获取超时流中的数据
        DataStream<String> timeoutStream = resultStream.getSideOutput(timeoutTag);

        // 正常流中的数据和超时流中的数据都是匹配的对象
        DataStream<String> totalStream = resultStream.union(timeoutStream);

        // 将数据写到kafka中
        totalStream.addSink(FlinkSinkUtil.getKafkaSink(Constants.TOPIC_DWD_TRAFFIC_UJ_DETAIL));
    }
}
